package com.ibm.cps.join;

import java.util.Collection;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.newmessage.BasicDataSourceMetadata;
import com.ibm.cps.newmessage.MetadataInputSchema;
import com.ibm.datamodel.timeseries.TimeSeriesSchema;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

public abstract class AbstractJoiner {
	
	public ITimeSeriesSchema schema;
	public BasicDataSourceMetadata dsmetadata;
	
	public ITimeSeriesSchema getSchema() {
		return schema;
	}
	
	public AbstractJoiner(String metadata) throws CPSException{
		MetadataInputSchema inputSchema = getInputSchema();
		if (inputSchema == null) {
			throw new CPSException(ErrorCode.INTERNAL_ERROR_CODE, "No publisher metadata schema is defined.");
		}
		
		
		ObjectNode node = JSONUtility.fromObjectString(metadata);
		inputSchema.validateInput(node);
		
		// get the simple schema and transform into full qualify-name
		dsmetadata = new BasicDataSourceMetadata(JSONUtility.fromObjectString(metadata));
		ITimeSeriesSchema simpleschema = dsmetadata.getTimeSeriesSchema();
		
		String dsName = dsmetadata.getName();
		ArrayNode tags = simpleschema.getTags().deepCopy();
		for(JsonNode tag : tags){
			ObjectNode field = (ObjectNode)tag;
			field.put(MessageFields.NAME, dsName + "." + field.get(MessageFields.NAME).asText());
		}
		schema =  new TimeSeriesSchema(dsmetadata.getTenantid(), dsmetadata.getSchemaid(), tags, 1);
	}
	
	
	public abstract Collection<ParsedDataMessage> query(ObjectNode node) throws CPSException;
	
	public abstract void close() throws CPSException;
	
	// define which field should in the joiner
	public abstract MetadataInputSchema getInputSchema() throws CPSException;

	
}
